package top.jolyoulu.core.example

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @Author: JolyouLu
 * @Date: 2024/2/6 16:32
 * @Description
 * 编号	字段名称	字段类型	字段含义
 * 1	date	String	用户点击行为的日期
 * 2	user_id	Long	用户的 ID
 * 3	session_id	String	Session 的 ID
 * 4	page_id	Long	某个页面的 ID
 * 5	action_time	String	动作的时间点
 * 6	search_keyword	String	用户搜索的关键词
 * 7	click_category_id	Long	某一个商品品类的 ID
 * 8	click_product_id	Long	某一个商品的 ID
 * 9	order_category_ids	String	一次订单中所有品类的 ID 集合
 * 10	order_product_ids	String	一次订单中所有商品的 ID 集合
 * 11	pay_category_ids	String	一次支付中所有品类的 ID 集合
 * 12	pay_product_ids	String	一次支付中所有商品的 ID 集合
 * 13	city_id	Long	城市 id
 *
 * 存在问题
 * Q1.actionRDD被多次使用
 * Q2.cogroup可能会shuffle
 */
object Spark01_Example_HotCategoryTop10Analysis_Optimize1 {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10Analysis")
    val sc: SparkContext = new SparkContext(sparkConf)

    //读取日志数据
    val path: String = this.getClass.getClassLoader.getResource("datas/user_visit_action.txt").toURI.getPath
    val actionRDD: RDD[String] = sc.textFile(path)
    actionRDD.cache() //Q1 => 使用cache

    //统计品类点击数量(品类ID,点击数量)
    //过滤符合条件的数据
    val clickActionRDD: RDD[String] = actionRDD.filter(
      action => {
        val datas: Array[String] = action.split("_")
        //品类ID和产品ID不是-1就是 品类
        datas(6) != "-1"
      }
    )
    //处理过滤后数据，组合成(品类ID,点击数量)
    val clickCountRDD: RDD[(String, Int)] = clickActionRDD.map(
      action => {
        val datas: Array[String] = action.split("_")
        (datas(6), 1)
      }
    ).reduceByKey(_ + _)

    //统计品类下单数量(品类ID,下单数量)
    //过滤符合条件的数据
    val orderActionRDD: RDD[String] = actionRDD.filter(
      action => {
        val datas: Array[String] = action.split("_")
        //一次订单中所有品类的 ID 集合非null就是 下单
        datas(8) != "null"
      }
    )
    //处理过滤的数据，将[15,1,20,6,4] 拆分成 (品类ID,下单数量)
    val orderCountRDD: RDD[(String, Int)] = orderActionRDD.flatMap(action => {
      val datas: Array[String] = action.split("_")
      val cid = datas(8)
      val cids = cid.split(",")
      cids.map(id => (id, 1))
    }).reduceByKey(_ + _)

    //统计品类支付数量(品类ID,支付数量)
    //过滤符合条件的数据
    val payActionRDD: RDD[String] = actionRDD.filter(
      action => {
        val datas: Array[String] = action.split("_")
        //一次支付中所有品类的 ID 集合非null就是 下单
        datas(10) != "null"
      }
    )
    //处理过滤的数据，将[15,1,20,6,4] 拆分成 (品类ID,支付数量)
    val payCountRDD: RDD[(String, Int)] = payActionRDD.flatMap(action => {
      val datas: Array[String] = action.split("_")
      val cid = datas(10)
      val cids = cid.split(",")
      cids.map(id => (id, 1))
    }).reduceByKey(_ + _)

    //将品类进行排序，并且取前10
    //点击数量排序，下单数量排序，支付数量排序
    //利用元组排序：比较第一个，比较第二个，比较第三个 (品类ID,(点击数量,下单数量,支付数量))

    //变化结构(品类ID,下单数量) => (品类ID,(点击数量,0,0))
    val rdd1: RDD[(String, (Int, Int, Int))] = clickCountRDD.map {
      case (cid, cnt) => {
        (cid, (cnt, 0, 0))
      }
    }
    //变化结构(品类ID,下单数量) => (品类ID,(0,下单数量,0))
    val rdd2: RDD[(String, (Int, Int, Int))] = orderCountRDD.map {
      case (cid, cnt) => {
        (cid, (0, cnt, 0))
      }
    }
    //变化结构(品类ID,下单数量) => (品类ID,(0,0,支付数量))
    val rdd3: RDD[(String, (Int, Int, Int))] = payCountRDD.map {
      case (cid, cnt) => {
        (cid, (0, 0, cnt))
      }
    }
    //将三个数据源合并一起，进行聚合计算
    val sourceRDD: RDD[(String, (Int, Int, Int))] = rdd1.union(rdd2).union(rdd3)
    val analysisRDD: RDD[(String, (Int, Int, Int))] = sourceRDD.reduceByKey(
      (t1, t2) => {
        (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
      }
    )

    val resultRDD: Array[(String, (Int, Int, Int))] = analysisRDD.sortBy(_._2, false).take(10)

    //将结果采集到控制台打印
    resultRDD.foreach(println)

    //关闭环境
    sc.stop()
  }
}
